package com.hm.kafkatest2.service;

import com.hm.kafkatest2.entity.KafkaMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Service
public class KafkaListenerService {

    private final ConsumerFactory<String, String> consumerFactory;
    private final KafkaMessageService kafkaMessageService;
    private final Map<String, ConcurrentMessageListenerContainer<String, String>> containers = new ConcurrentHashMap<>();
    
    // 定义包含数字和字母的字符集
    private static final String CHARACTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";

    public KafkaListenerService(ConsumerFactory<String, String> consumerFactory, KafkaMessageService kafkaMessageService) {
        this.consumerFactory = consumerFactory;
        this.kafkaMessageService = kafkaMessageService;
    }
    
    /**
     * 生成8位包含数字和字母的唯一标识
     * 
     * @return 8位随机字符串
     */
    private String generateUniqueId() {
        Random random = new Random();
        StringBuilder sb = new StringBuilder(8);
        for (int i = 0; i < 8; i++) {
            sb.append(CHARACTERS.charAt(random.nextInt(CHARACTERS.length())));
        }
        return sb.toString();
    }

    /**
     * 检测Kafka集群连接状态
     *
     * @param bootstrapServers Kafka集群地址
     * @return 连接检测结果，如果连接成功返回null，否则返回错误信息
     */
    public String testConnection(String bootstrapServers) {
        org.apache.kafka.clients.consumer.Consumer<String, String> consumer = null;
        try {
            // 创建消费者配置
            Map<String, Object> consumerConfigs = new ConcurrentHashMap<>(consumerFactory.getConfigurationProperties());
            consumerConfigs.put("bootstrap.servers", bootstrapServers);
            consumerConfigs.put("group.id", "connection-test-" + System.currentTimeMillis());
            
            // 设置较短的连接超时时间，避免长时间等待
            consumerConfigs.put("connections.max.idle.ms", 3000);
            consumerConfigs.put("request.timeout.ms", 3000);
            consumerConfigs.put("session.timeout.ms", 3000);
            consumerConfigs.put("heartbeat.interval.ms", 1000);
            consumerConfigs.put("metadata.fetch.timeout.ms", 3000);
            consumerConfigs.put("default.api.timeout.ms", 3000);
            
            // 创建消费者工厂和消费者实例
            ConsumerFactory<String, String> customConsumerFactory = new DefaultKafkaConsumerFactory<>(consumerConfigs);
            consumer = customConsumerFactory.createConsumer();
            
            // 尝试列出主题来测试连接
            consumer.listTopics();
            
            log.info("成功连接到Kafka集群: {}", bootstrapServers);
            return null; // 返回null表示连接成功
        } catch (Exception e) {
            String errorMessage = "无法连接到Kafka集群: " + bootstrapServers + 
                    "，请检查集群地址和网络连接。错误信息: " + e.getMessage();
            log.error("无法连接到Kafka集群: {}", bootstrapServers, e);
            return errorMessage; // 返回错误信息
        } finally {
            // 关闭消费者资源
            if (consumer != null) {
                try {
                    consumer.close(java.time.Duration.ofSeconds(1));
                } catch (Exception e) {
                    log.warn("关闭Kafka消费者时出现异常", e);
                }
            }
        }
    }

    /**
     * 开启对指定Kafka集群和topic的监听
     *
     * @param bootstrapServers Kafka集群地址
     * @param topic            要监听的topic名称
     * @param groupId          消费组ID
     * @return 监听是否成功启动
     */
    public boolean startListening(String bootstrapServers, String topic, String groupId) {
        // 为每个监听器生成唯一标识，支持重复参数的多个监听器
        String uniqueId = generateUniqueId();
        
        // 使用传入的groupId或生成默认的groupId
        String consumerGroupId = (groupId != null && !groupId.isEmpty()) ? groupId : 
            "kafka-test-group-" + System.currentTimeMillis();
        
        String containerKey = bootstrapServers + "%" + topic + "%" + consumerGroupId + "%" + uniqueId;
        
        try {
            // 创建消费者配置
            Map<String, Object> consumerConfigs = new ConcurrentHashMap<>(consumerFactory.getConfigurationProperties());
            consumerConfigs.put("bootstrap.servers", bootstrapServers);
            
            consumerConfigs.put("group.id", consumerGroupId);

            // 创建消费者工厂
            ConsumerFactory<String, String> customConsumerFactory = new DefaultKafkaConsumerFactory<>(consumerConfigs);

            // 创建容器属性
            ContainerProperties containerProperties = new ContainerProperties(topic);

            // 设置消息监听器
            containerProperties.setMessageListener((MessageListener<String, String>) data -> {
//                log.info("监听到消息 - Kafka集群: {}, Topic: {}, 分区: {}, Offset: {}, 消费组: {}, 消息内容: {}",
//                        bootstrapServers, topic, data.partition(), data.offset(), consumerGroupId, data.value());
                
                // 将消息存储到数据库
                KafkaMessage kafkaMessage = new KafkaMessage();
                kafkaMessage.setBootstrapServers(bootstrapServers);
                kafkaMessage.setTopic(topic);
                kafkaMessage.setGroupId(consumerGroupId);
                kafkaMessage.setPartition(data.partition());
                kafkaMessage.setOffset(data.offset());
                kafkaMessage.setMessage(data.value());
                kafkaMessage.setUniqueId(uniqueId); // 添加唯一标识
                kafkaMessage.setCreateTime(LocalDateTime.now());
                
                try {
                    kafkaMessageService.save(kafkaMessage);
                    log.info("{}:{}的消息已成功存储到数据库:{}",bootstrapServers, topic,data.offset());
                } catch (Exception e) {
                    log.error("存储消息到数据库失败", e);
                }
            });

            // 创建并启动监听容器
            ConcurrentMessageListenerContainer<String, String> container =
                    new ConcurrentMessageListenerContainer<>(customConsumerFactory, containerProperties);
            container.setBeanName("kafka-listener-" + uniqueId);
            container.start();

            // 存储容器引用
            containers.put(containerKey, container);

            log.info("成功启动对Kafka集群 {} 的Topic {} 的监听，消费组: {}", bootstrapServers, topic, consumerGroupId);
            return true;
        } catch (Exception e) {
            log.error("启动监听失败 - Kafka集群: {}, Topic: {}, 消费组: {}", bootstrapServers, topic, groupId, e);
            return false;
        }
    }

    /**
     * 开启对指定Kafka集群和topic的监听（兼容旧方法）
     *
     * @param bootstrapServers Kafka集群地址
     * @param topic            要监听的topic名称
     * @return 监听是否成功启动
     */
    public boolean startListening(String bootstrapServers, String topic) {
        return startListening(bootstrapServers, topic, null);
    }

    /**
     * 停止对指定Kafka集群和topic的监听
     *
     * @param bootstrapServers Kafka集群地址
     * @param topic            要停止监听的topic名称
     * @param groupId          消费组ID
     * @return 监听是否成功停止
     */
    public boolean stopListening(String bootstrapServers, String topic, String groupId) {
        String consumerGroupId = (groupId != null && !groupId.isEmpty()) ? groupId : 
            "kafka-test-group-" + System.currentTimeMillis();
        
        // 查找匹配的监听器并停止
        boolean stopped = false;
        Iterator<Map.Entry<String, ConcurrentMessageListenerContainer<String, String>>> iterator = 
            containers.entrySet().iterator();
        
        while (iterator.hasNext()) {
            Map.Entry<String, ConcurrentMessageListenerContainer<String, String>> entry = iterator.next();
            String key = entry.getKey();
            // 解析key，检查是否匹配用户提供的参数
            String[] parts = key.split("%");
            if (parts.length >= 3 && 
                parts[0].equals(bootstrapServers) && 
                parts[1].equals(topic) && 
                parts[2].equals(consumerGroupId)) {
                
                try {
                    entry.getValue().stop();
                    iterator.remove(); // 使用迭代器安全地移除元素
                    log.info("成功停止对Kafka集群 {} 的Topic {} 的监听，消费组: {}", bootstrapServers, topic, consumerGroupId);
                    stopped = true;
                    // 注意：这里只停止第一个匹配的监听器，如果要停止所有匹配的监听器，需要移除break语句
                    break;
                } catch (Exception e) {
                    log.error("停止监听失败 - Kafka集群: {}, Topic: {}, 消费组: {}", bootstrapServers, topic, consumerGroupId, e);
                    return false;
                }
            }
        }
        
        if (!stopped) {
            log.warn("未找到对Kafka集群 {} 的Topic {} 的监听容器，消费组: {}", bootstrapServers, topic, consumerGroupId);
        }
        
        return stopped;
    }

    /**
     * 根据唯一标识停止特定的监听器
     *
     * @param bootstrapServers Kafka集群地址
     * @param topic            要停止监听的topic名称
     * @param groupId          消费组ID
     * @param uniqueId         唯一标识
     * @return 监听是否成功停止
     */
    public boolean stopListeningById(String bootstrapServers, String topic, String groupId, String uniqueId) {
        String consumerGroupId = (groupId != null && !groupId.isEmpty()) ? groupId : 
            "kafka-test-group-" + System.currentTimeMillis();
        
        String containerKey = bootstrapServers + "%" + topic + "%" + consumerGroupId + "%" + uniqueId;
        log.info("正在停止:{}",containerKey);
        ConcurrentMessageListenerContainer<String, String> container = containers.get(containerKey);
        if (container != null) {
            try {
                container.stop();
                containers.remove(containerKey);
                log.info("成功停止对Kafka集群 {} 的Topic {} 的监听，消费组: {}，唯一标识: {}", 
                        bootstrapServers, topic, consumerGroupId, uniqueId);
                return true;
            } catch (Exception e) {
                log.error("停止监听失败 - Kafka集群: {}, Topic: {}, 消费组: {}，唯一标识: {}", 
                        bootstrapServers, topic, consumerGroupId, uniqueId, e);
                return false;
            }
        } else {
            log.warn("未找到对Kafka集群 {} 的Topic {} 的监听容器，消费组: {}，唯一标识: {}", 
                    bootstrapServers, topic, consumerGroupId, uniqueId);
            return false;
        }
    }

    /**
     * 停止所有具有相同参数的监听器
     *
     * @param bootstrapServers Kafka集群地址
     * @param topic            要停止监听的topic名称
     * @param groupId          消费组ID
     * @return 停止的监听器数量
     */
    public int stopAllListening(String bootstrapServers, String topic, String groupId) {
        String consumerGroupId = (groupId != null && !groupId.isEmpty()) ? groupId : 
            "kafka-test-group-" + System.currentTimeMillis();
        
        int stoppedCount = 0;
        Iterator<Map.Entry<String, ConcurrentMessageListenerContainer<String, String>>> iterator = 
            containers.entrySet().iterator();
        
        while (iterator.hasNext()) {
            Map.Entry<String, ConcurrentMessageListenerContainer<String, String>> entry = iterator.next();
            String key = entry.getKey();
            // 解析key，检查是否匹配用户提供的参数
            String[] parts = key.split("%");
            if (parts.length >= 3 && 
                parts[0].equals(bootstrapServers) && 
                parts[1].equals(topic) && 
                parts[2].equals(consumerGroupId)) {
                
                try {
                    entry.getValue().stop();
                    iterator.remove(); // 使用迭代器安全地移除元素
                    log.info("成功停止对Kafka集群 {} 的Topic {} 的监听，消费组: {}", bootstrapServers, topic, consumerGroupId);
                    stoppedCount++;
                } catch (Exception e) {
                    log.error("停止监听失败 - Kafka集群: {}, Topic: {}, 消费组: {}", bootstrapServers, topic, consumerGroupId, e);
                }
            }
        }
        
        if (stoppedCount == 0) {
            log.warn("未找到对Kafka集群 {} 的Topic {} 的监听容器，消费组: {}", bootstrapServers, topic, consumerGroupId);
        } else {
            log.info("成功停止 {} 个对Kafka集群 {} 的Topic {} 的监听容器，消费组: {}", 
                    stoppedCount, bootstrapServers, topic, consumerGroupId);
        }
        
        return stoppedCount;
    }

    /**
     * 停止对指定Kafka集群和topic的监听（兼容旧方法）
     *
     * @param bootstrapServers Kafka集群地址
     * @param topic            要停止监听的topic名称
     * @return 监听是否成功停止
     */
    public boolean stopListening(String bootstrapServers, String topic) {
        return stopListening(bootstrapServers, topic, null);
    }

    /**
     * 获取监听状态
     *
     * @return 当前所有监听的状态信息
     */
    public Map<String, Boolean> getListeningStatus() {
        Map<String, Boolean> statusMap = new ConcurrentHashMap<>();
        containers.forEach((key, container) -> {
            // 解析key以提供更友好的状态显示
            String[] parts = key.split("%");
            if (parts.length >= 4) {
                // 包含唯一ID的完整key，格式为: bootstrapServers%topic%groupId%uniqueId
                String displayKey = parts[0] + "%" + parts[1] + "%" + parts[2] + "%" + parts[3];
                log.info("正在监听:{}",displayKey);
                statusMap.put(displayKey, container.isRunning());
            } else {
                statusMap.put(key, container.isRunning());
            }
        });
        return statusMap;
    }
}